package com.shujia.hbase;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;
import org.junit.Before;
import org.junit.Test;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.List;

public class Demo3Dianxin {

    private Connection connection;

    @Before
    public void startup() {
        Configuration configuration = new Configuration();

        //指定zk的连接地址
        configuration.set("hbase.zookeeper.quorum", "master:2181,node1:2181,node2:2181");

        //创建连接
        try {
            connection = ConnectionFactory.createConnection(configuration);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Test
    public void saveData() throws IOException {
        Table dianxin = connection.getTable(TableName.valueOf("dianxin"));


        BufferedReader bufferedReader = new BufferedReader(new FileReader("data/DIANXIN.csv"));

        String line;

        while ((line = bufferedReader.readLine()) != null) {

            String[] split = line.split(",");
            //手机号
            String mdn = split[0];

            //数据发送的时间
            String time = split[1];


            //经纬度
            String x = split[4];
            String y = split[5];

            String value = x + "," + y;


            Put put = new Put(mdn.getBytes());
            put.addColumn("info".getBytes(), "xy".getBytes(), Long.parseLong(time), value.getBytes());

            //插入一条数据
            dianxin.put(put);
        }

    }

    @Test
    public void getLastLocation() throws Exception {
        Table dianxin = connection.getTable(TableName.valueOf("dianxin"));

        Get get = new Get("99ED5EAED77722DEA0D2216CC7CF4A0C7B960756".getBytes());

        //指定需要查询版本的数量
        get.setMaxVersions(10);

        Result result = dianxin.get(get);

        //获取多条
        List<Cell> cells = result.listCells();

        for (Cell cell : cells) {
            String xy = Bytes.toString(CellUtil.cloneValue(cell));

            System.out.println(xy);
        }


    }
}
